package edu.corn.job.core.engine;

import edu.corn.common.design.chain.Chain;
import edu.corn.common.idmng.SnowflakeUtil;
import edu.corn.job.config.JobConfigProperties;
import edu.corn.job.core.context.JobRunningContext;
import edu.corn.job.core.engine.mutex.MutexPool;
import edu.corn.job.core.processor.CallFallback;
import edu.corn.job.core.statemachine.JobEventEnum;
import edu.corn.job.core.statemachine.JobStateEnum;
import edu.corn.job.core.statemachine.JobStateUtil;
import edu.corn.job.core.statemachine.JobStatemachineFactory;
import edu.corn.statemachine.core.StateMachine;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.util.HashMap;
import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import static edu.corn.job.core.statemachine.JobStateUtil.running;

/**
 * 抽象jobEngine，在该层规定提交方法。
 * <p>
 * 提供几个能力。
 * 创建线程池。
 * 呈递任务。
 * 异步和同步的，有回调的，没有回调的。
 */
@Data
@Slf4j
public abstract class JobEngine {

    private String jobEngineName;

    /**
     * 等待被取走呈递的任务队列。
     */
    protected Queue<JobRunningContext> waitingList = new LinkedList<>();

    /**
     * 暂时挂起的任务队列。
     */
    protected Queue<JobRunningContext> pendingList = new LinkedList<>();

    /**
     * 当前正在执行的job，可以被强制终止。
     * id-jobRunner
     */
    protected HashMap<String, JobRunner> runningSet = new HashMap<>();

    /**
     * 状态机持有类。
     */
    protected StateMachine<JobStateEnum, JobEventEnum> statemachine;

    protected JobConfigProperties config;

    public enum State {
        NEW,
        RUNNING,
        PHASE,
        SHUTDOWN,
        ;
    }

    protected State STATE;


    /**
     * 锁池。引擎需要有并发访问的能力。
     */
    protected MutexPool mutexPool;

    /**
     * 外部传进来的已经装配好的责任链。
     */
    protected HashMap<String, Chain<JobRunningContext>> chains;


    public abstract ExecutorService getExecutorService();


    /**
     * 确保引擎已经可提供服务状态。
     *
     * @return
     */
    public Boolean ensuringState() {
        return this.STATE == State.RUNNING;
    }

    /**
     * 工作线程lock
     */
    private volatile ReentrantLock lock;

    private Condition workerLock;
    private Condition engineLock;

    public JobEngine() {
        this.STATE = State.NEW;
    }

    void lock() {
        try {
            engineLock.await();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    void unlock() {
        engineLock.signal();
    }

    /**
     * 提交后执行，考虑在哪能让子类自动执行。
     */
    public void notifyWorker() {
        workerLock.notify();
    }


    /**
     * 提交任务，会经过校验链。
     *
     * @param context
     * @param callFallback
     */
    public final void submit(JobRunningContext context, CallFallback callFallback) {
        // 验证引擎状态
        if (!ensuringState()) {
            try {
                engineLock.await();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }

        for (Chain<JobRunningContext> chain : chains.values()) {
            boolean execute = chain.execute(context);
            if (!execute) {
                // 记录未通过
                return;
            }
        }

        context.setCallable(callFallback);
        JobStateUtil.submit(context);
    }

    /**
     * 没有回调的提交任务。
     *
     * @param context
     */
    public final void submit(JobRunningContext context) {
        submit(context, null);
    }


    /**
     * 直接提交的任务。
     */
    final void submitJob(JobRunner jobRunner) {
        // before submit
        JobRunningContext context = jobRunner.getContext();
        if (context.getMasterId() == null) {
            context.putMasterId(SnowflakeUtil.generateIdStr());
        }

        // do
        doSubmit(jobRunner);

        // after submit
    }

    /**
     * 将具体何时，什么方式呈递交给子类。
     *
     * @param jobRunner runner
     */
    abstract void doSubmit(JobRunner jobRunner);

    /**
     * 将任务挂起。
     *
     * @param jobContext
     */
    public void pendingJob(JobRunningContext jobContext) {
        // TODO 挂起的时候也需要释放全部的锁，然后继续执行。
        JobStateUtil.pending(jobContext);
    }

    public void terminateJob(JobRunningContext jobContext) {
        // 停止任务。
    }


    /**
     * 引擎拥有查找是否存在相同任务的功能。
     * - 可以同时开启多个任务，但是要由用户确定是否开启。
     * -
     *
     * @param jobContext
     * @return 是否重复。
     */
    public boolean isDuplicateJob(JobRunningContext jobContext) {
        String masterId = jobContext.getMasterId();
        if (masterId == null) {
            return false;
        }

        lock();
        boolean isDup = false;
        try {
            // 这里尽可能让引擎不暂停。如果有并发问题，再考虑解决方案。
            // TODO 这时候这几个集合应该是不可变的。先这样。
            for (JobRunningContext context : waitingList) {
                if (context.getMasterId().equals(masterId)) {
                    isDup = true;
                }
            }
            for (JobRunningContext context : pendingList) {
                if (context.getMasterId().equals(masterId)) {
                    isDup = true;
                }
            }
            if (runningSet.containsKey(masterId)) {
                isDup = true;
            }
        } finally {
            unlock();
        }
        return isDup;
    }


    /**
     * 初始化jobEngine 内部数据结构。
     */
    public void init(MutexPool mutexPool, String jobEngineName, HashMap<String, Chain<JobRunningContext>> chains,JobConfigProperties config) {
        log.info("jobEngine: {} init", jobEngineName);
        if (lock == null) {
            synchronized (JobEngine.class) {
                if (lock == null) {
                    lock = new ReentrantLock();
                    workerLock = lock.newCondition();
                    engineLock = lock.newCondition();
                }
            }
        } else {
            return;
        }
        // 初始化状态机
        this.statemachine = JobStatemachineFactory.getStateMachine(this);
        this.mutexPool = mutexPool;
        this.jobEngineName = jobEngineName;
        this.chains = chains;
        this.config = config;
        log.info("jobEngine: {} init ending", jobEngineName);
    }

    public void start() {
        /**
         * 启动工作线程。
         */
        Worker worker = new Worker();
        worker.run();
        this.STATE = State.RUNNING;
        log.info("jobEngine: {} start", jobEngineName);
    }

    /**
     * 从持久化层面加载job。
     */
    public void load() {
        log.info("jobEngine: {} load", jobEngineName);

    }

    /**
     * 将job持久化。
     */
    public void save() {
        log.info("jobEngine: {} save", jobEngineName);

    }

    /**
     * 停止jobEngine
     */
    public void shutdown() {
        log.info("jobEngine: {} shutdown", jobEngineName);
    }


    private final class Worker implements Runnable {

        /**
         * worker线程不仅要负责递交任务，还负责检查线程队列，如果有线程死锁超过多少分钟了，将其杀死并且告警。
         */
        @Override
        public void run() {
            while (true) {
                while (waitingList.isEmpty()) {
                    try {
                        workerLock.wait(); // 等待被唤醒
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        return;
                    }

                    if (!ensuringState()) {
                        try {
                            engineLock.await();
                        } catch (InterruptedException e) {
                            throw new RuntimeException(e);
                        }
                    }

                    JobRunningContext jobContext = waitingList.poll();
                    // 从waiting list取出任务并提交线程池执行。
                    if (jobContext != null) {
                        running(jobContext);
                    }
                }
            }
        }
    }
}
